package org.fujene.communicate.indexer;

import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.zip.GZIPInputStream;
import java.util.zip.GZIPOutputStream;

import org.fujene.communicate.lock.IndexableSwitch;
import org.fujene.constants.Constants;
import org.fujene.memman.ReplicateMemoryManager;
import org.fujene.structure.TermIndexee;
import org.fujene.toolbox.ByteConverter;

public class TermIndexerClient implements Runnable{
	private static final int TERMFREQ = Constants.BYTE_TERMFREQ;
	private static final int STRLEN = Constants.BYTE_STRLEN;

	private static ReplicateMemoryManager memman;
	private static LinkedBlockingQueue<TermIndexee> clientQueue;
	private static Object clientBlockade;
	private static GZIPOutputStream clientLog;
	private static File logFile;

	static{
		clientQueue = new LinkedBlockingQueue<TermIndexee>();
		clientBlockade = new Object();
	}

	public TermIndexerClient(boolean isCreateLog){
		if(isCreateLog && logFile == null){
			logFile = new File(".term_clientlog");
			if(!logFile.exists()) try{
				logFile.createNewFile();
			}catch(IOException e){
				System.err.println("Client-log file does not create. Indexing works without logging.");
			}
			try{
				clientLog = new GZIPOutputStream(new FileOutputStream(logFile));
			}catch(FileNotFoundException e){}catch(IOException e){}
		}
	}

	private static void addLog(TermIndexee newIndex) throws IOException{
		byte[] writebytes;
		clientLog.write(ByteConverter.longToByte(newIndex.getID(), 8));
		clientLog.write(ByteConverter.longToByte(newIndex.getTermFreq(), TERMFREQ));
		clientLog.write(ByteConverter.longToByte(newIndex.getValueType(), 1));
		writebytes = newIndex.getTerm().getBytes("UTF-8");
		clientLog.write(ByteConverter.intToByte(writebytes.length, STRLEN));
		clientLog.write(writebytes);
	}

	public static void revert() throws FileNotFoundException, IOException{
		GZIPInputStream logstream = null;
		try{
			logstream = new GZIPInputStream(new FileInputStream(logFile));
			byte[] strb;
			long ID;
			int termfreq, valuetype;
			while(logstream.available() != 0){
				strb = new byte[8];
				logstream.read(strb);
				ID = ByteConverter.byteToLong(strb);
				strb = new byte[TERMFREQ];
				logstream.read(strb);
				termfreq = ByteConverter.byteToInt(strb);
				strb = new byte[1];
				logstream.read(strb);
				valuetype = ByteConverter.byteToInt(strb);

				strb = new byte[2];
				logstream.read(strb);
				strb = new byte[ByteConverter.byteToInt(strb)];
				logstream.read(strb);

				clientQueue.add(TermIndexee.asValue(new String(strb, "UTF-8"), ID, termfreq, valuetype));
			}
		}finally{
			if(logstream != null) try{
				logstream.close();
			}catch(IOException e){}
		}
	}

	// Clear log file and re-construct
	public static void clearLog() throws IOException{
		if(clientLog != null){
			clientLog.close();
			logFile.delete();
			logFile.createNewFile();
			clientLog = new GZIPOutputStream(new FileOutputStream(logFile));
		}
	}

	public static void property(ReplicateMemoryManager manager){
		memman = manager;
	}

	public static void addToQueue(TermIndexee newIndex){
		if(clientLog != null) try{
			addLog(newIndex);
		}catch(IOException e){}
		clientQueue.add(newIndex);
	}

	public static boolean isEmpty(){
		return clientQueue.isEmpty();
	}

	public static void waitUntilEmpty(){
		if(!clientQueue.isEmpty()) synchronized(clientBlockade){
			try{
				clientBlockade.wait();
			}catch(InterruptedException e){}
		}
	}

	public static int residue(){
		return clientQueue.size();
	}

	@Override
	public void run(){
		TermIndexee newIndex = null;
		while(true){
			try{
				newIndex = clientQueue.take();
				if(clientQueue.isEmpty()) synchronized(clientBlockade){
					clientBlockade.notifyAll();
				}
			}catch(InterruptedException e1){}
			try{
				IndexableSwitch.waitIfLocked();
			}catch(InterruptedException e){}
			if(newIndex == null) break;
			memman.storeTerm(newIndex.getTerm(), newIndex.getID(), newIndex.getTermFreq(),
					newIndex.getValueType());
		}
	}
}
